package cn.itcast.flink.window.time;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/**
* 窗口统计案例演示：滑动计数窗口（Sliding Count Window)，数字累加求和统计
*/
public class StreamSlidingCountWindow_4 {
public static void main(String[] args) throws Exception {
// 1. 执行环境-env：流计算执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1) ;
// 2. 数据源-source：Socket接收数据
DataStreamSource<String> inputStream = env.socketTextStream("node1.itcast.cn", 9999);
// 3. 转换处理-transformation：调用DataStream函数，处理数据
    SingleOutputStreamOperator<Integer> numStream = inputStream
            .filter(line -> null != line && line.trim().length() > 0)
            .map(line -> Integer.parseInt(line.trim()));
    //TODO 每隔5秒累加
    SingleOutputStreamOperator<Integer> sumStream = numStream.countWindowAll(10, 5)
            .sum(0);
// 4. 数据终端-sink
        sumStream.printToErr();
// 5. 触发执行-execute
env.execute(StreamSlidingCountWindow_4.class.getSimpleName()) ; } }